package com.kd.crawler.main;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Properties;
import java.util.Queue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.jms.Message;
import javax.jms.TextMessage;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSON;
import com.kd.crawler.common.jms.ActiveMQConsumer;
import com.kd.crawler.common.jms.ActiveMQConsumer.ReceiveListener;
import com.kd.crawler.common.jms.ActiveMQProducer;
import com.kd.crawler.common.utils.PropertiesUtils;
import com.kd.crawler.crawl.MQMessageHashSender;
import com.kd.crawler.crawl.Spider;
import com.kd.crawler.crawl.Spider.SpiderListener;
import com.kd.crawler.entity.CrawlerEntry;
import com.kd.crawler.parser.thread.MQMessageSender;

/**
 * @author Manfred
 *
 */
public class CrawlMain {
	
	private Logger logger = Logger.getLogger(CrawlMain.class);

	private Queue<Spider> spiderQueue = new LinkedList<Spider>();
	private static HashMap<Integer,Long> crawlTimeMap=new HashMap<Integer,Long>();
	
	public void start() {
		final Properties crawlPro = PropertiesUtils.getProperties("crawl.properties");
		final Properties MQPro = PropertiesUtils.getProperties("activeMQ.properties");
		String threadCount = crawlPro.getProperty("thread_count");
		int defaultThreadCount = 1;
		if (threadCount != null) {
			defaultThreadCount = Integer.valueOf(threadCount);
		}
		final ExecutorService pool = Executors.newFixedThreadPool(defaultThreadCount);
		for (int i = 0; i < defaultThreadCount; i++) {
			offerSpider(new Spider(new SpiderListener() {
				public void onFinish(Spider spider) {
//					crawlTimeMap.put(spider.getCrawlerUrl().getJobID(), System.currentTimeMillis());
					offerSpider(spider);
				}
			}));
		}
		while (true) {
			try {
				ActiveMQConsumer.recevieQueueMessage(MQPro.getProperty("QUEUE_CRAWL_ENTRY_MESSAGE"), new ReceiveListener() {
					public void onReceive(Message message) {
						if (message == null) {
							return;
						}
						String messageText;
						try {
							messageText = ((TextMessage) message).getText();
							CrawlerEntry cu=JSON.parseObject(messageText, CrawlerEntry.class);
							Long lastCrawlTime=crawlTimeMap.get(cu.getJobID());

							//lastCrawlTime!=null&&(System.currentTimeMillis()/1000-lastCrawlTime)<600
							if(lastCrawlTime!=null){
								logger.debug(cu.getUrl()+" --> 已有代表");
//								logger.debug(cu.getUrl()+" --> 间隔时间未到");
//								ActiveMQProducer.sendQueueMessage(MQPro.getProperty("QUEUE_CRAWL_ENTRY_MESSAGE"), JSON.toJSONString(cu));

							if(lastCrawlTime!=null&&(System.currentTimeMillis()-lastCrawlTime)<cu.getIntervalTime()){
								logger.debug(cu.getUrl()+" --> 间隔时间未到");
//								ActiveMQProducer.sendQueueMessage(MQPro.getProperty("QUEUE_CRAWL_ENTRY_MESSAGE_NEW"), JSON.toJSONString(cu));
								MQMessageHashSender.getInstance().offer("QUEUE_CRAWL_ENTRY_MESSAGE", 0, messageText, cu.getUrl());

							}else{
								logger.info("从队列接收到入口信息:" + messageText);
								Spider spider = pollSpider();
								spider.setCrawlerUrl(cu);
								pool.execute(spider);
								crawlTimeMap.put(cu.getJobID(), System.currentTimeMillis()/1000);
							}
							message.acknowledge();
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
				});
			} catch (Exception e) {
				logger.error("接收activeMessage报错:", e);
			}
		}
	}
	private synchronized Spider pollSpider() {
		Spider spider = null;
		while (spider == null) {
			spider = spiderQueue.poll();
			if (spider == null) {
				try {
					this.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		return spider;
	}
//	private synchronized void offerSpider(Spider spider) {
//		spiderQueue.offer(spider);
//		this.notify();
//	}
//
//	/**
//	 * @param args
//	 */
//	public static void main(String[] args) {
//		new CrawlMain().start();
//	}

}
}
